#include "NetIO.h"

Epoll_Reactor::Epoll_Reactor( void )
{
#ifndef WIN32
	epoll_fd_ = epoll_create( EPOLL_SIZE );
	epoll_event ev = { 0, { 0 } };
	ev.events = EPOLLIN | EPOLLERR | EPOLLHUP;
	ev.data.fd = interrupter_.read_handle( );
	epoll_ctl( epoll_fd_, EPOLL_CTL_ADD, interrupter_.read_handle( ), &ev );
#endif
}

Epoll_Reactor::~Epoll_Reactor( void )
{
}

void Epoll_Reactor::cancel_handle( Socket handle )
{
	Handle_Data *data = handle.handle_data( );
	if( handle == INVALID_HANDLE || data == 0 )
		return;

	Operation_Queue op_queue;
	Guard< Synch_Mutex > handle_guard( data->mutex_ );
	if ( data->is_cancel_ )
		return;

	for( int i = 0; i < OperationType::CONNECT; i ++ )
	{
		while( Operation *op = data->operations_[ i ].front( ) )
		{
			op->errcode( IOCANCELED );
			op_queue.push( op );
			data->operations_[ i ].pop( );
		}
	}
	data->is_cancel_ = true;
	handle_guard.release( );
	
	{
		Guard< Synch_Mutex > guard( mutex_ );
		complete_operations_.concat( &op_queue );
	}

	{
		Guard< Synch_Mutex > guard_map( handle_map_mutex_ );
		handle_map_.erase( handle );
	}

#ifndef WIN32
	epoll_event ev = { 0, { 0 } };
	epoll_ctl( epoll_fd_, EPOLL_CTL_DEL, handle, &ev );
#endif
}

int Epoll_Reactor::start_op( Socket &handle, Operation *op )
{
	int op_type = op->optype( );
	if( op_type == OperationType::CONNECT )
		op_type = OperationType::WRITE;
	else if( op_type == OperationType::ACCEPT )
		op_type = OperationType::READ;

	Guard< Synch_Mutex > guard( handle.handle_data( )->mutex_ );
	if( handle.handle_data( )->is_cancel_ )
		return -1;

	if( handle.handle_data( )->operations_[ op_type ].empty( ) )
	{
		if( op->perform( ) )
		{
			complete_operations_.push( op );
			return 0;
		}
	}

	handle.handle_data( )->operations_[ op_type ].push( op );
	return 0;
}

int Epoll_Reactor::run_once( )
{
	Guard< Synch_Mutex > guard( mutex_ );
	handle_all_complete_op( );
	guard.release( );

	if( exit_ )
		return -1;
	Operation_Queue op_queue;
#ifndef WIN32
	epoll_event events[ 128 ];
	static const int flag[ 2 ] = { EPOLLIN, EPOLLOUT };
	int num_events = epoll_wait( epoll_fd_, events, 128, 5000 );

	for( int i = 0; i < num_events; i ++ )
	{
		if( events[ i ].data.fd == interrupter_.read_handle( ) )
		{
			interrupter_.reset( );
			continue;
		}

		Handle_Data *handle_data_ptr = events[ i ].data.ptr;
		Guard< Mutex >  guard_handle( handle_data_ptr->mutex_ );
		if( handle_data_ptr->is_cancel_ )
			continue;

		for( int j = OperationType::WRITE; j >= 0; --j )
		{
			if( events[ i ].events & ( flag[ j ] | EPOLLERR | EPOLLHUP ) )
			{
				Operation* op;
				while( op = handle_data_ptr->operations_[ j ].front( ) )
				{
					if( !op->perform( ) )
						break;

					handle_data_ptr->operations_[ j ].pop( );
					op_queue.push( op );
				}
			}
		}
	}
#endif

	guard.acquire( );
	complete_operations_.concat( &op_queue );
	handle_all_complete_op( );
	
	return 0;
}

int Epoll_Reactor::register_handler_i( Socket &handle )
{
#ifndef WIN32
	epoll_event ev = { 0, { 0 } };
	ev.events = EPOLLIN | EPOLLERR | EPOLLHUP | EPOLLOUT | EPOLLPRI | EPOLLET;
	ev.data.ptr = handler.event_data( );
	if( epoll_ctl( epoll_fd_, EPOLL_CTL_ADD, handle.handle( ), &ev ) != 0 )
		return errno;
#endif
	return 0;
}

